package com.moli.iov.handle.real.mq.kafka;

import javax.annotation.Resource;


import com.alibaba.fastjson.JSONObject;
import com.moli.iov.contant.RealContant;
import com.moli.iov.handle.real.service.IVehicleService;
import lombok.Data;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;

import com.moli.iov.cache.RedisUtil;
import com.moli.iov.consumer.KafkaConsumerServer;
import com.moli.iov.handle.real.util.Contact;
import com.moli.iov.protocol.model.dto.VehicleOnLineDto;
import com.moli.iov.util.GsonUtil;

@Data
public class OnlineConsumer extends KafkaConsumerServer{

	@Resource
	IVehicleService vehicleService;

	@Resource
	RedisUtil<String, String> redisUtil;

	@Override
	public void manualHandle(ConsumerRecord<String, Object> record,
			Acknowledgment acknowledgment) {

		VehicleOnLineDto dto;
		long start = System.currentTimeMillis();
		try {
			dto = (VehicleOnLineDto)record.value();
			if(dto!=null){
				logger.info("[车辆上下线][vin:{},status:{},time:{}]",dto.getVin(),dto.getStatus(),dto.getTimestamp());
				if(dto.getStatus()){
					redisUtil.hset(RealContant.KEY_ONLINE, dto.getVin(),dto.getVin());
					redisUtil.hset(Contact.KEY_ONLINE_TIME,dto.getVin(),dto.getVin()+"_"+System.currentTimeMillis());
					vehicleService.onlineNotice(dto);
				}else{
					redisUtil.hdel(RealContant.KEY_ONLINE, dto.getVin());
				}
			}else{
				logger.error("[车辆上下线][key:{}][values is null]",record.key());
			}
			acknowledgment.acknowledge();
		} catch (Exception e) {
			logger.error("[车辆上下线][处理异常][key:{},value:{}][{}]",record.key(),record.value(),e.getMessage());
			try{
				acknowledgment.acknowledge();
			}catch (Exception submitEx){
				submitEx.printStackTrace();
			}
		}

		long time = System.currentTimeMillis() - start;
		logger.info("[车辆:{}，上下线:{}][时长:{}ms]",record.key(),record.value(),time);
	}
	
}
